package com.chenjj.bigdata.flink.connectors.kafka;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.junit.Test;

import java.io.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

public class KafkaConnectorTester {

    Properties properties = new Properties();

    {
        properties.setProperty("bootstrap.servers", "192.168.152.135:9092");
    }

    /**
     * 测试生产数据
     *
     * 验证： bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic flink-demo --from-beginning
     */
    @Test
    public void testProduce() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> stream = env.addSource(new FileDataSource("D:\\Code\\Gitee\\flink-demo\\src\\main\\resources\\deposit_trans3.txt"));

        FlinkKafkaProducer<String> producer = new FlinkKafkaProducer("flink-demo",new SimpleStringSchema(),properties);

        stream.addSink(producer);

        env.execute("testProcuder");
    }




    /**
     *  以文件作为数据源，每秒推送一条数据到Flink
     */
    public class FileDataSource implements SourceFunction<String> {
        volatile boolean isRunning = true;
        private  String filePath;
        private  long interval=1;

        private FileDataSource(){}

        public FileDataSource(String filePath){
            this.filePath = filePath;
        }

        public FileDataSource(String filePath,long interval){
            this.filePath = filePath;
            this.interval = interval;
        }

        public void run(SourceContext ctx) throws Exception {
            List<String> lines = readFile02(filePath);
            int i = 0;
            while (isRunning){
                if (i < lines.size()){
                    String line = lines.get(i);
                    ctx.collect(line);
                    i++;
                }
                Thread.sleep(interval);
            }
        }

        public void cancel() {
            isRunning = false;
        }

        /**
         * 读取一个文本 一行一行读取
         *
         * @param path
         * @return
         * @throws IOException
         */
        public List<String> readFile02(String path)  {
            // 使用一个字符串集合来存储文本中的路径 ，也可用String []数组
            List<String> list = new ArrayList<String>();
            FileInputStream fis = null;
            InputStreamReader isr = null;
            BufferedReader br = null;
            try {
                fis = new FileInputStream(path);
                isr = new InputStreamReader(fis, "UTF-8");
                br = new BufferedReader(isr);
                String line = "";
                while ((line = br.readLine()) != null) {
                    list.add(line);
                }
            } catch (FileNotFoundException e) {
                e.printStackTrace();
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                try {
                    if (br != null)  br.close();
                    if (isr != null) isr.close();
                    if (fis != null) fis.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

            return list;
        }
    }

}
